/*
 * HandlerAdapter.java    0.0.1    18/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.server.impl;


import java.io.IOException;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.concurrent.Callable;

import org.mentalsmash.tazio.net.cs.server.ProtocolHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The HandlerAdapter has the role of adapting ProtocolHandlers to the multiplexed IO infrastructure,
 * in particular to the {@link org.mentalsmash.tazio.net.cs.server.impl.Dispatcher Dispatcher} interface.
 * <p>
 * It abstracts one of the channels the dispatcher is listening to, and also takes care of running
 * the action associated with that channel's readiness event.
 * </p> 
 * <p>
 * This class and all the classes in this package are largely inspired to the presentation
 * "How To Build A Scalable NIO Server, Mark II" by Ron Hitchens (http://ronsoft.com)
 * </p>
 * @author     Ron Hitchens - original work  
 * @author     Andrea Reale <andrea.reale@gmail.com> - adaptation
 * @version    0.0.1    18/set/2009
 */
class HandlerAdapter implements Callable<HandlerAdapter>, ChannelFacade {
	/*
	 * This implementation is for the most taken from Ron Hitchen's implementation
	 * with slight modifications from Andrea Reale.
	 */
	
	/*
	 * SYNCHORNIZATION: more then one thread may access an HandlerAdapter in the case that
	 * two or more readiness events are fired by the dispatcher while some other preceding
	 * readiness event is still being processed. Therefore every action that changes the state
	 * of this channel has to be synchronzied
	 */
	
	
	private final static Logger log = LoggerFactory.getLogger(HandlerAdapter.class);
	private final LowLevelDispatcher dispatcher;
	private final InputQueue inputQueue; 
	private final OutputQueue outputQueue;
	private final Object stateChangeLock = new Object(); // this lock is used to synchronize threads
	                                                     // who wants to act upon channel related
	                                                     // structures
	
	private ProtocolHandler clientHandler; //the application handler that will be called upon readiness
	                                       //events
	
	private SelectionKey key = null;
	private SelectableChannel channel = null;
	
	private volatile int interestOps = 0; //operations on the channel the applications is interested in
	                                      // (byte mask)
	private int readyOps = 0;
	private boolean shuttingDown = false;
	private volatile boolean running = false; //true if this handler is currently being run by some thread
	private volatile boolean dead = false; //indicates wether the handler has encountered any exception
	                                       //and therefore has not completed his job. (It is dead)
	
	
	

	/**
	 * Create a new <code>HandlerAdapter</code> given the specified parameters.
	 * @param clientHandler the <code>ProtocolHandler</code> that will be called 
	 * 						when incoming bytes arrive
	 * @param dispatcher the dispatcher this handler is associated to
	 * @param pool the pool of direct buffers used to implement input and output queues
	 */
	public HandlerAdapter (ProtocolHandler clientHandler, LowLevelDispatcher dispatcher,
		BufferPool pool) {
		this.dispatcher = dispatcher;
		this.clientHandler = clientHandler;

		inputQueue = new ByteBufferInputQueue(pool);
		outputQueue = new ByteBufferOutputQueue(pool, this);
	}

	// ------------------------------------------------------------
	// Implementation of Callable<HandlerAdapter> interface


	/**
	 * Implementation of the {@link java.util.concurrent.Callable Callable} interface for the
	 * handler adaptor. That's what runs when a readiness event for the associated channel is
	 * fired by the dispatcher.
	 * 
	 * @return the Handler adaptor itself
	 * @throws IOException if something goes wrong while accessing the channel 
	 *                     (both in input or output)
	 */
	@Override
	public HandlerAdapter call() throws IOException {

		try {
			if(channel instanceof WritableByteChannel) {
				drainOutput(); //If theres data waiting to be sent, send it out.
				log.debug("" + this + " has drained its output.");
			} else {
				log.debug(this + " hasn't drained its output: the channel is not writable.");
			}
			
			if(channel instanceof ReadableByteChannel) {
				fillInput(); // Now read, if there's somthing to read
				log.debug("" + this + " has filled its input.");
			} else {
				log.debug(this + " hasn't filled its input: the channel is not readable.");
			}
			
			ByteBuffer message;

			// must process all buffered messages because Selector will
			// not fire again for input that's already read and buffered
			while ((message = clientHandler.nextMessage (this)) != null) {
				clientHandler.handleMessage (message, this);
				//notifies the dispatcher
				dispatcher.requesteServed();
				log.debug("" + this + "has sent a complete message to its protocol handler.");
			}
		} finally {
			synchronized (stateChangeLock) {
				running = false;
				log.debug(""+this+" has completed its exectuion.");
			}
		}

		return this;
	}

	// ------------------------------------------------------------
	// Implementation of ChannelFacade interface

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.ChannelFacade#getInputQueue
	 */
	@Override
	public InputQueue getInputQueue() {
		return inputQueue;
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.ChannelFacade#getOutputQueue
	 */
	@Override
	public OutputQueue getOutputQueue() {
		return outputQueue;
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.ChannelFacade#modifyInterestsOps(int,int)
	 */
	@Override
	public void modifyInterestOps (int opsToSet, int opsToReset) {
		synchronized (stateChangeLock) {
			int oldInterests = interestOps; //DEBUG
			
			interestOps = (interestOps | opsToSet) & (~opsToReset);
			log.debug("Interest ops for " +this + " modified from " + oldInterests + " to " + interestOps);
			if (!running) {
				dispatcher.enqueueStatusChange (this);
			}
		}
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.ChannelFacade#getInterestsOps
	 */
	@Override
	public int getInterestOps() {
		return (interestOps);
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.ChannelFacade#setProtocolHandler(ProtocolHandler)
	 */
	@Override
	public void setProtocolHandler (ProtocolHandler handler) {
		if(handler==null){
			log.error("Handler cannot be null");
			throw new IllegalArgumentException("ProtocolHandler cannto be null");
		}
		clientHandler = handler;
	}
	
	/**
	 * Return the protocol handlers associated to this HandlerAdapter
	 * @return the protocol handlers associated to this HandlerAdapter
	 */
	public ProtocolHandler getProtocolHandler() {
		return clientHandler;
	}

	// --------------------------------------------------
	// package-local methods called by the Dispatcher

	/**
	 * Before running the handler we have to update this handler's status variables to
	 * reflect the very last changes made to it. In particular the set of current interesting events,
	 * the set of current readiness events, and the handler's running status have to be changed.
	 */
	void prepareToRun() {
		if(key == null){
			String msg = "Unexpected error: handler should have a registration key from a dispatcher" +
				"before running";
			log.error(msg);
			throw new LowLevelDispatchingException(msg);
		}
		
		/* 
		 * This method should be accessed only by the Dispatcher's thread. Nevertheless it needs to
		 * be synchronised wrt other possibly running threads that have access to the handler's status
		 */
		synchronized (stateChangeLock) {
			interestOps = key.interestOps();
			readyOps = key.readyOps();
			running = true;
		}
	}

	/**
	 * This method is called by the Dispatcher's thread after this handler's associated channel
	 * gets registered on it. It saves a reference to the registration key in the handler, and updates
	 * channel and interestOps operations
	 * @param key the selection key, referring to the registration on the Dispatcher of the channel
	 *            associated to this handler
	 */
	void setKey(SelectionKey key) {
		this.key = key;
		channel = key.channel();
		interestOps = key.interestOps();
	}

	/**
	 * Returns the selection key referring to the registration of this handler's channel on its
	 * dispatcher. 
	 * @return the selection key referring to the registration of this handler's channel on its
	 *         dispatcher.
	 */
	SelectionKey key() {
		return key;
	}

	/**
	 * Checks whether this instance of the handler has died during is last execution.
	 * An handler is said to be dead if some kind of exeption is raised during its protocol handler
	 * execution
	 * @return true if this instance of the handler has died during is last execution.
	 */
	boolean isDead() {
		return dead;
	}

	/**
	 * Cause the channel to be de-registered from the Selector upon return, and sets this
	 * instance as dead
	 */
	void die() {
		dead = true;
		log.debug(this + " has died.");
	}

	// --------------------------------------------------
	// package-local life-cycle helper methods, called by the Dispatcher

	// Called when registering, but before the handler is active
	void registering() {
		log.debug(this + " is registering.");
//		clientHandler. (this);
	}

	/**
	 * Called when the handler is registered, but before the first message
	 */
	void registered() {
		clientHandler.registered(this);
	}

	/**
	 * Called when unregistration has been requested, but while the
	 * handler is still active and able to interact with the framework.
	 * Extension Point: This implementation simply calls through to
	 * the client handler, which may or may not be running.  Either
	 * the client code must take steps to protect its internal state,
	 * or logic could be added here to wait until the handler finishes.
	 */
	void unregistering() {
		log.debug(this + " is unregistering");
//		clientHandler.stopping (this);
	}


	/**
	 * Called when the handler has been unregistered and is no longer active.
 	 * If unregistering() waits for the handler to finish, then this
	 * one should be safe.  If not, then this function has the same
	 * concurrency concerns as does unregistering().
	 */
	void unregistered() {
		clientHandler.unregistered (this);
	}

	// --------------------------------------------------
	// Private helper methods

	// These three methods manipulate the private copy of the selection
	// interest flags.  Upon completion, this local copy will be copied
	// back to the SelectionKey as the new interest set.
	private void enableWriteSelection() {
		modifyInterestOps (SelectionKey.OP_WRITE, 0);
	}

	private void disableWriteSelection() {
		modifyInterestOps (0, SelectionKey.OP_WRITE);
	}

	private void disableReadSelection() {
		modifyInterestOps (0, SelectionKey.OP_READ);
	}


	/**
	 * If there is output queued, and the channel is ready to
	 * accept data, send as much as it will take.
	 */
	private void drainOutput() throws IOException {
		if (((readyOps & SelectionKey.OP_WRITE) != 0)
		    && ( ! outputQueue.isEmpty()))
		{
			outputQueue.drainTo ((WritableByteChannel) channel);
		}

		// Write selection is turned on when output data in enqueued,
		// turn it off when the queue becomes empty.
		if (outputQueue.isEmpty()) {
			disableWriteSelection();

			if (shuttingDown) {
				channel.close();
				clientHandler.unregistered (this);
			}
		}
	}


	/**
	 * Attempt to fill the input queue with as much data as the channel
	 * can provide right now.  If end-of-stream is reached, stop read
	 * selection and shutdown the input side of the channel.
	 * @throws IOException if there's an error while reading from the channel
	 */
	private void fillInput() throws IOException {
		if (shuttingDown) return;

		int rc = inputQueue.fillFrom ((ReadableByteChannel) channel);

		if (rc == -1) { //End of stream
			disableReadSelection();

			if (channel instanceof SocketChannel) {
				SocketChannel sc = (SocketChannel) channel;

				if (sc.socket().isConnected()) {
					try {
						sc.socket().shutdownInput();
					} catch (SocketException e) {
						// happens sometimes, ignore
					}
				}
			}

			shuttingDown = true;
//			clientHandler.stopping(this);

			// cause drainOutput to run, which will close
			// the socket if/when the output queue is empty
			if(channel instanceof WritableByteChannel) {
				enableWriteSelection();
			}
		}
	}
	
	/**
	 *{@inheritDoc} 
	 */
	@Override
	public String toString(){
		return "HandlerAdapter for " + this.clientHandler + ".";
	}
	
	/**
	 *{@inheritDoc} 
	 */
	@Override
	public boolean isReadable(){
		return channel instanceof ReadableByteChannel;
	}
	
	/**
	 *{@inheritDoc} 
	 */
	@Override
	public boolean isWritable() {
		return channel instanceof WritableByteChannel;
	}
	
	/*
	 * Following: package-level methods that should be called only for testing purposes
	 */
	
	SelectionKey getSelectionKey() { 
		return this.key();
	}
	
}
